package com.ld.security.common.rocketmq.mq;

import com.ld.security.common.core.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 消费者
 */
@Slf4j
public class SuperConsumer<E> extends    SuperMq<E>  implements MessageListenerConcurrently {


    public SuperConsumer(String producerGroup, String topic, String subExpression) {
        super(producerGroup, topic, subExpression);
    }

    public SuperConsumer(SuperMq mq) {
        super(mq);
    }

    @Override
    public void run(E msg) {

    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt message : msgs) {
            String json = new String(message.getBody());
            runBefore(json);
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    public void runBefore(String json){
        log.info("消费消息"+json);
        E v = (E) JsonUtil.toObject(json,this.getEntityClass());
        run(v);
    }

}
